/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.news;

import cn.ac.iie.di.global.GlobalParas;
import static cn.ac.iie.di.global.GlobalParas.PARSER_POOL;
import cn.ac.iie.di.rocketmq.Message;
import cn.ac.iie.di.server.ZZPTParserStartup;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.NoSuchElementException;
import java.util.Random;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.log4j.Logger;

/**
 *
 * @author lenovo-pc
 */
public class JsonNewsSender implements Runnable {

    private int addCount = 0;
    private long startTime = System.currentTimeMillis();
    static private final Logger logger = Logger.getLogger(JsonNewsSender.class.getName());

    ThreadPoolExecutor threadpool;

    public JsonNewsSender() {
        //创建一个线程池
        threadpool = new ThreadPoolExecutor(GlobalParas.PARSER_POOL - 12, GlobalParas.PARSER_POOL, 3, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(GlobalParas.QUEUE_SIZE), new ThreadFactory() {
//        threadpool = new ThreadPoolExecutor(16, 32, 3, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(20), new ThreadFactory() {
            AtomicInteger idx = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "JsonNewsTask_" + idx.getAndIncrement());
            }
        }, new RejectedExecutionHandler() {

            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                while (true) {
                    try {
                        executor.getQueue().put(r);
                        break;
                    } catch (InterruptedException ex) {
                        logger.warn("", ex);
                    }
                }
            }
        });
    }

    @Override
    public void run() {

        String json = null;
        ArrayList<Message> jsonList;
        while (true) {
            jsonList = new ArrayList<>(10);
            try {

                for (int i = 0; i < 10; ++i) {
                    Message str = GlobalParas.PULL_MQ_QUEUE.poll(500, TimeUnit.MILLISECONDS);
                    
//                    测试数据用
//                    Message str = ZZPTParserStartup.offer();
                    if (str == null) {
                        break;
                    }
                    jsonList.add(str);
                }
                if (jsonList.isEmpty()) {
                    Thread.sleep(5000);
                    logger.info("There isn't any data in the mq...");
                    continue;
                }
//                logger.info("xxxxxxxx");
                logger.info("data is prepared...");
                threadpool.execute(new JsonNewsTask(jsonList));
                ++addCount;
            } catch (Exception ex) {
                logger.error("this fans thread error :" + ex, ex);
                ex.printStackTrace();
            }
        }

    }

}
